/**
 * File: ParallelBatchController.java
 * Created by: mhaimel
 * Created on: 23 Jul 2009
 * CVS:  $Id: ParallelBatchController.java,v 1.2 2009/09/01 10:59:59 mhaimel Exp $
 */
package uk.ac.ebi.curtain.controller;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicReference;

import uk.ac.ebi.curtain.management.IProcessor;
import uk.ac.ebi.curtain.management.InterceptorInfo;

/**
 * @author mhaimel
 *
 */
public abstract class ParallelBatchController<I extends Serializable, O extends Serializable, P extends IProcessor<I, O>> extends ParallelController<I, O, P> {
	private static final Integer BATCH_SIZE = 1000;
	private AtomicReference<List<I>> list = 
		new AtomicReference<List<I>>(
			Collections.synchronizedList(new ArrayList<I>()));
	private Integer batchSize;
	
	public ParallelBatchController() {
		this(BATCH_SIZE);
	}

	public ParallelBatchController(Integer batchSize) {
		this.batchSize = batchSize;
	}
	
	
	protected P buildJob(I input, uk.ac.ebi.curtain.management.InterceptorInfo info) {
		addInput(input);
		_submitBatch(false);
		return null;
	}

	protected void addInput(I input) {
		this.list.get().add(input);
	};
	
	protected void _submitBatch(boolean force) {
		if(force || getBatchSize()<= list.get().size()){
			List<I> currList = list.getAndSet(Collections.synchronizedList(new ArrayList<I>()));
			_doSubmit(currList);
		}
	}
	
	protected abstract void _doSubmit(List<I> currList);
	
	@Override
	public void shutDown(boolean force, InterceptorInfo info) {
		if(!force){
			_submitBatch(true);
		}
		super.shutDown(force, info);
	}
	
	@Override
	protected void processResult(Future<O> resJob,P job) {
		Collection<O> resColl = extractBatchResult(resJob, job);
		if(null != resColl){
			for(O out : resColl){
				interceptorHandle(out);
			}
		}
	}
	
	protected Collection<O> extractBatchResult(Future<O> res, P job){
		return (Collection<O>) extractResult(res, job);
	}
	
	@Override
	protected void doShutdown() {
		if(!list.get().isEmpty()){
			getLog().error("Forgot to process " + list.get().size());
		}
		super.doShutdown();
	}

	public void setBatchSize(Integer batchSize) {
		this.batchSize = batchSize;
	}

	protected Integer getBatchSize() {
		return batchSize;
	}	
}
